/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.apache.james.mailbox.cassandra.user;

/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/


import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.nio.ByteBuffer;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;


import org.apache.cassandra.thrift.TBinaryProtocol;

import org.apache.james.mailbox.SubscriptionException;

import org.apache.james.mailbox.cassandra.CNames;
import org.apache.james.mailbox.cassandra.CNonTransactionalMapper;
import org.apache.james.mailbox.store.user.SubscriptionMapper;
import org.apache.james.mailbox.store.user.model.Subscription;
import org.apache.james.mailbox.store.user.model.impl.SimpleSubscription;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

/**
 * HBase implementation of a {@link SubscriptionMapper}. 
 * I don't know if this class is thread-safe!
 * 
 */
public class CSubscriptionMapper extends CNonTransactionalMapper implements SubscriptionMapper {

  
    /*
     * (non-Javadoc)
     * @see org.apache.james.mailbox.store.user.SubscriptionMapper#findMailboxSubscriptionForUser(java.lang.String, java.lang.String)
     */
         
    @Override
    public Subscription findMailboxSubscriptionForUser(String user, String mailbox) throws SubscriptionException {
        
        
        
        try {
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         transport.open();
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);    
        
         ColumnParent subscriptions = new ColumnParent(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
         Subscription subscription = null;
         ColumnPath c =new ColumnPath();
         c.setColumn(CNames.SUBCRIPTION_MAILBOX);
         c.setColumn_family(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
            
            ColumnOrSuperColumn k= client.get(ByteBufferUtil.bytes(user), c, ConsistencyLevel.ONE);
            byte[] b=k.getColumn().getValue();
            
                if (k.getColumn().isSetValue()) 
                {
                    //****************
                    
                   if((b.toString()).equalsIgnoreCase(mailbox))
                   {
                    subscription = new SimpleSubscription(user, mailbox);
                    return subscription;
                    }
       
                }
           
        } catch (TTransportException ex) {
                Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (NotFoundException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new SubscriptionException(e);
        } 
         return null;
    }

    /**
     * @throws SubscriptionException 
     * @see org.apache.james.mailbox.store.user.SubscriptionMapper#save(Subscription)
     */
    
    @Override
    public void save(Subscription subscription) throws SubscriptionException {
        //TODO: maybe switch to checkAndPut
        //HTable subscriptions = null;
        
        try {
         TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
         TProtocol protocol = new TBinaryProtocol(transport);
         transport.open();
         Cassandra.Client client = new Cassandra.Client(protocol);
         client.set_keyspace(CNames.keyspace);   
         
         
           ColumnParent subscriptions = new ColumnParent(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
           String mlbx = subscription.getMailbox();
           String user = subscription.getUser();
           
            Column newcolumn = new Column(CNames.SUBCRIPTION_MAILBOX);
            
            newcolumn.setValue(ByteBufferUtil.bytes(mlbx));
            newcolumn.setTimestamp(System.currentTimeMillis());
            
            client.insert(ByteBufferUtil.bytes(user), subscriptions,newcolumn,ConsistencyLevel.ONE) ;
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new SubscriptionException(e);
        } 
    }

    /**
     * @throws SubscriptionException 
     * @see org.apache.james.mailbox.store.user.SubscriptionMapper#findSubscriptionsForUser(java.lang.String)
     */
    @Override
    public List<Subscription> findSubscriptionsForUser(String user) throws SubscriptionException {
        try {
            TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
            TProtocol protocol = new TBinaryProtocol(transport);
            transport.open();
          
            Cassandra.Client client = new Cassandra.Client(protocol);
            client.set_keyspace(CNames.keyspace);   
            
            ColumnParent subscriptions = new ColumnParent(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
            List<Subscription> subscriptionList = new ArrayList<Subscription>();
            SlicePredicate predicate = new SlicePredicate();
            predicate.setSlice_range(new SliceRange(ByteBuffer.wrap(new byte[0]), ByteBuffer.wrap(new byte[0]), false, 100));
            List<ColumnOrSuperColumn> columnsByKey = client.get_slice(ByteBufferUtil.bytes(user), subscriptions, predicate, ConsistencyLevel.ALL);
            if (columnsByKey.isEmpty()) {
                 for (ColumnOrSuperColumn c : columnsByKey) {
                     
                     //************
                    subscriptionList.add(new SimpleSubscription(user,(c.getColumn().getValue()).toString()));
                    
                                       
                }
                return subscriptionList;
            } 
            
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new SubscriptionException(e);
        } 
        return null;
    }

    /**
     * @throws SubscriptionException 
     * @see org.apache.james.mailbox.store.user.SubscriptionMapper#delete(Subscription)
     */
    @Override
    public void delete(Subscription subscription) throws SubscriptionException {
        //TODO: maybe switch to checkAndDelete
        
        try {
            
           TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
           TProtocol protocol = new TBinaryProtocol(transport);
           transport.open();
           Cassandra.Client client = new Cassandra.Client(protocol);
           client.set_keyspace(CNames.keyspace);  
            
           ColumnPath c=new ColumnPath();
           c.setColumn_family(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
           c.setColumn(CNames.SUBCRIPTION_MAILBOX);
           ColumnOrSuperColumn d = new ColumnOrSuperColumn();
           d= client.get(ByteBufferUtil.bytes(subscription.getUser()), c, ConsistencyLevel.ONE);
           client.remove(ByteBufferUtil.bytes(subscription.getUser()), c, d.getColumn().getTimestamp(), ConsistencyLevel.ONE);
        } catch (NotFoundException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvalidRequestException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnavailableException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TimedOutException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (TException ex) {
            Logger.getLogger(CSubscriptionMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IOException e) {
            throw new SubscriptionException(e);
        } 
    }
}
